package com.atguigu.sparkstreaming.demos

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/5/23
 *
 *    获取offsets：  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
 *
 *    注意事项:
 *                offsetRanges 在Driver端获取。
 *
 *                只有在算子中编写的运算逻辑，才是在Executor端运行，其他部分全部在Driver端。
 *
 *                只有RDD是KafkaRDD类型，才能调用rdd.asInstanceOf[HasOffsetRanges].offsetRanges。
 *                只有初始DS，才是DirectKafkaInputDStream 里面封装的才是KafkaRDD，应该调用初始DS的foreachRDD或transform才能获取到偏移量
 *
 *
 * Exception in thread "main" java.lang.ClassCastException:
 *    org.apache.spark.rdd.MapPartitionsRDD
 *        cannot be cast to
 *   org.apache.spark.streaming.kafka010.HasOffsetRanges
 *
 *
 */
object GetOffsetsDemo2 {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "SparkStreamingKafkaDemo", Seconds(5))


    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "atguigu1227",
      "auto.offset.reset" -> "latest",
      // ①取消offsets的自动提交
      "enable.auto.commit" -> "false"
    )


    val topics1 = Array("topicA")


    // DirectKafkaInputDStream 里面封装的才是KafkaRDD
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics1, kafkaParams)
    )

    val ds1: DStream[String] = stream.map(record => record.value())


    stream.foreachRDD(rdd => {

      /*
           获取到的当前批次数据的偏移量

           OffsetRange: 一个分区的偏移量信息。

           OffsetRange(topic: 'topicA', partition: 2, range: [3 -> 3])
              关注:  untilOffset，才是要提交的位置

       */
     //Driver端:  当前线程:streaming-job-executor-0
     println("当前线程:"+Thread.currentThread().getName)

     // 没有调用RDD的任何方法
      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      offsetRanges.foreach(offsetRange => println(offsetRange))

     // 调用了rdd的foreach算子
      rdd.foreach(record => {
        //Executor端运行 : Executor task launch worker for task 27
        //println(record.value()+":"+Thread.currentThread().getName)
        println(record+":"+Thread.currentThread().getName)
      })

    })

    // 启动app
    streamingContext.start()

    // 阻塞当前线程，让程序24h不停运行
    streamingContext.awaitTermination()


  }

}
